/*
 * Copyright 2011-2014 the original author or authors.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.dianwoba.universe.redis.operation.impl;

import java.util.concurrent.TimeUnit;

import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.TimeoutUtils;

import com.dianwoba.universe.cache.JedisCallback;
import com.dianwoba.universe.cache.JedisTemplate;
import com.dianwoba.universe.redis.operation.ValueOperations;
import com.fasterxml.jackson.core.type.TypeReference;

import redis.clients.jedis.BinaryJedisCommands;

/**
 * Default implementation of {@link ValueOperations}.
 * 
 */
public class DefaultValueOperations<K, V, X extends BinaryJedisCommands> 
    extends AbstractOperations<K, V, X> implements ValueOperations<K, V> {

	public DefaultValueOperations(JedisTemplate<K, V, X> template) {
		super(template);
	}

	public V get(final Object key) {

		return execute(new ValueDeserializingRedisCallback(key) {

			protected byte[] inRedis(byte[] rawKey, X jedis) {
				return jedis.get(rawKey);
			}
		});
	}
	
	@SuppressWarnings("unchecked")
	@Override
	public <T> T get(Object key, Class<T> clazz) {
		return (T)execute(new ValueDeserializingRedisCallback(key, clazz) {

			protected byte[] inRedis(byte[] rawKey, X jedis) {
				return jedis.get(rawKey);
			}
		});
	}
	
	@SuppressWarnings("unchecked")
	@Override
	public <T> T get(Object key, TypeReference<?> valueTypeRef) {
		return (T)execute(new ValueDeserializingRedisCallback(key, valueTypeRef) {

			protected byte[] inRedis(byte[] rawKey, X jedis) {
				return jedis.get(rawKey);
			}
		});
	}
	
	public V getAndSet(K key, V newValue) {
		final byte[] rawValue = rawValue(newValue);
		return execute(new ValueDeserializingRedisCallback(key) {

			protected byte[] inRedis(byte[] rawKey, X jedis) {
				return jedis.getSet(rawKey, rawValue);
			}
		});
	}
	
	@SuppressWarnings("unchecked")
	@Override
	public <T> T getAndSet(K key, V newValue, Class<T> clazz) {
		final byte[] rawValue = rawValue(newValue);
		return (T)execute(new ValueDeserializingRedisCallback(key, clazz) {

			protected byte[] inRedis(byte[] rawKey, X jedis) {
				return jedis.getSet(rawKey, rawValue);
			}
		});
	}
	
	@SuppressWarnings("unchecked")
	@Override
	public <T> T getAndSet(K key, V value, TypeReference<?> valueTypeRef) {
		return (T)execute(new ValueDeserializingRedisCallback(key, valueTypeRef) {

			protected byte[] inRedis(byte[] rawKey, X jedis) {
				return jedis.get(rawKey);
			}
		});
	}

	public Long increment(K key, final long delta) {
		final byte[] rawKey = rawKey(key);
		return execute(new JedisCallback<Long, X>() {

			public Long doInRedis(X jedis) {
				return jedis.incrBy(rawKey, delta);
			}
		});
	}

	public Double increment(K key, final double delta) {
		final byte[] rawKey = rawKey(key);
		return execute(new JedisCallback<Double, X>() {
			public Double doInRedis(X jedis) {
				return jedis.incrByFloat(rawKey, delta);
			}
		});
	}

	public Integer append(K key, String value) {
		final byte[] rawKey = rawKey(key);
		final byte[] rawString = rawString(value);

		return execute(new JedisCallback<Integer, X>() {

			public Integer doInRedis(X jedis) {
				final Long result = jedis.append(rawKey, rawString); 				
				return ( result != null ) ? result.intValue() : null; 
			}
		});
	}

	public String get(K key, final long start, final long end) {
		final byte[] rawKey = rawKey(key);

		byte[] rawReturn = execute(new JedisCallback<byte[], X>() {

			public byte[] doInRedis(X jedis) {
				return jedis.getrange(rawKey, start, end);
			}
		});

		return deserializeString(rawReturn);
	}

	public void set(K key, V value) {
		final byte[] rawValue = rawValue(value);
		execute(new ValueDeserializingRedisCallback(key) {

			protected byte[] inRedis(byte[] rawKey, X jedis) {
				jedis.set(rawKey, rawValue);
				return null;
			}
		});
	}

	public void set(K key, V value, final long timeout, final TimeUnit unit) {
		final byte[] rawKey = rawKey(key);
		final byte[] rawValue = rawValue(value);

		execute(new JedisCallback<Object, X>() {

			public Object doInRedis(X jedis) throws DataAccessException {

				potentiallyUsePsetEx(jedis);
				return null;
			}

			public void potentiallyUsePsetEx(X jedis) {

				if (!TimeUnit.MILLISECONDS.equals(unit) || !failsafeInvokePsetEx(jedis)) {
					jedis.setex(rawKey, (int)TimeoutUtils.toSeconds(timeout, unit), rawValue);
				}
			}

			private boolean failsafeInvokePsetEx(X jedis) {

				boolean failed = false;
				try {
					jedis.setex(rawKey, (int)timeout, rawValue);
				} catch (UnsupportedOperationException e) {
					// in case the jedis does not support pSetEx return false to allow fallback to other operation.
					failed = true;
				}
				return !failed;
			}

		});
	}

	public Long setIfAbsent(K key, V value) {
		final byte[] rawKey = rawKey(key);
		final byte[] rawValue = rawValue(value);

		return execute(new JedisCallback<Long, X>() {

			public Long doInRedis(X jedis) throws DataAccessException {
				return jedis.setnx(rawKey, rawValue);
			}
		});
	}

	public void set(K key, final V value, final long offset) {
		final byte[] rawKey = rawKey(key);
		final byte[] rawValue = rawValue(value);

		execute(new JedisCallback<Object, X>() {

			public Object doInRedis(X jedis) {
				jedis.setrange(rawKey, offset, rawValue);
				return null;
			}
		});
	}

    public Boolean exists(K key) {
        final byte[] rawKey = rawKey(key);

        return execute(new JedisCallback<Boolean, X>() {

            public Boolean doInRedis(X jedis) throws DataAccessException {
                return jedis.exists(rawKey);
            }
        });
    }

	public Long size(K key) {
		final byte[] rawKey = rawKey(key);

		return execute(new JedisCallback<Long, X>() {

			public Long doInRedis(X jedis) {
				return jedis.strlen(rawKey);
			}
		});
	}
	
	@Override
	public Boolean setBit(K key, final long offset, final boolean value) {
		
		final byte[] rawKey = rawKey(key);
		return execute(new JedisCallback<Boolean, X>() {

			public Boolean doInRedis(X jedis) {
				return jedis.setbit(rawKey, offset, value);
			}
		});
	}

	@Override
	public Boolean getBit(K key, final long offset) {
		
		final byte[] rawKey = rawKey(key);
		return execute(new JedisCallback<Boolean, X>() {

			public Boolean doInRedis(X jedis) {
				return jedis.getbit(rawKey, offset);
			}
		});
	}

}
